from pyspark import SparkConf
from pyspark import SparkContext


def func_map(line):
    arr = line.split(',')
    print('执行map算子')
    return tuple(arr)


if __name__ == '__main__':
    conf = SparkConf().setMaster("local[1]").setAppName("spark01")
    sc = SparkContext(conf=conf)

    # 读取数据
    line_rdd = sc.textFile('../data/students.txt')
    # print(line_rdd.collect())

    # 把一行数据编程学生元祖
    student_tuple_rdd = line_rdd.map(func_map)
    # print(student_tuple_rdd.collect())

    # 当同一个RDD多次被使用的时候，或者同一个RDD在多个job中出现，可以将这个RDD做缓存
    student_tuple_rdd.cache() # 缓存
    # student_tuple_rdd.persist()

    # 根据学生的数据，统计每个班级的人数
    class_num_rdd = student_tuple_rdd.map(lambda student: (student[4], 1)).reduceByKey(lambda x, y: x + y)
    class_num_rdd.foreach(print)  # action算子

    # 根据学生的数据，统计每个性别的人数
    sex_num_rdd = student_tuple_rdd.map(lambda student: (student[3], 1)).reduceByKey(lambda x, y: x + y)
    sex_num_rdd.foreach(print)  # action算子
